﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Bit.Art.Core.Models;
using Bit.Art.Distributed;
using NetMQ;
using NetMQ.Sockets;

namespace Bit.Art.Api.Models
{
    internal class AsyncSocket : IDisposable
    {
        private readonly string m_serviceAddress;
        private NetMQPoller m_poller;
        private NetMQSocket m_dealerSocket;
        private ConcurrentDictionary<Guid, TaskCompletionSource<NetMQMessage>> m_requests;

        /// <summary>
        /// 
        /// </summary>
        public static AsyncSocket Instance { get; private set; } = new AsyncSocket(SiteConfiguration.AccountCheckEndpoint);

        /// <summary>
        /// 
        /// </summary>
        /// <param name="address"></param>
        internal AsyncSocket(string address)
        {
            m_serviceAddress = address;
            m_requests =
              new ConcurrentDictionary<Guid, TaskCompletionSource<NetMQMessage>>();
        }

        internal void Start()
        {
            m_dealerSocket = new DealerSocket();
            m_dealerSocket.ReceiveReady += OnReceiveReady;
            m_dealerSocket.Connect(m_serviceAddress);

            m_poller = new NetMQPoller();
            m_poller.Add(m_dealerSocket);

            Task.Factory.StartNew(() => m_poller.Run(),
              TaskCreationOptions.LongRunning);
        }

        internal Task<NetMQMessage> SendAndReceiveAsync(NetMQMessage message)
        {
            // duplicate the message because we are not the owner of the message
            NetMQMessage duplicteMessage = new NetMQMessage(message);

            var task = new Task<Task<NetMQMessage>>(() =>
            {
                var taskCompletionSource =
                  new TaskCompletionSource<NetMQMessage>();

                // because we are using a dealer we have to push the delimiter
                duplicteMessage.PushEmptyFrame();

                var requestid = Guid.NewGuid();
                // sending the request id the request identifier
                duplicteMessage.Push(requestid.ToByteArray());

                // add the request to the pending request dictionary
                m_requests.AddOrUpdate(requestid, taskCompletionSource, (key, oldvalue) => taskCompletionSource);

                m_dealerSocket.SendMultipartMessage(duplicteMessage);

                return taskCompletionSource.Task;
            });

            // will start the task on the scheduler which 
            // is the same thread as the Poller thread
            task.Start(m_poller);
            return task.Result;
        }

        private void OnReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            NetMQMessage message = m_dealerSocket.ReceiveMultipartMessage();

            // pop the request id
            var requestIdData = message.Pop().ToByteArray();
            var requestId = new Guid(requestIdData);

            // pop the delimiter
            message.Pop();

            TaskCompletionSource<NetMQMessage> taskCompletionSource;

            // getting the task completion source, if we were also try to 
            // handle timeout the request will be 
            // gone and the response will be dropped
            if (m_requests.TryGetValue(requestId, out taskCompletionSource))
            {
                taskCompletionSource.SetResult(message);
                // m_requests.Remove(requestId);
                m_requests.Remove(requestId, out TaskCompletionSource<NetMQMessage> value);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public void Dispose()
        {
            m_poller.Stop();
            m_dealerSocket.Dispose();
        }
    }
}
